package UDF

import Source.SensorReading
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.table.api.EnvironmentSettings
import org.apache.flink.table.api.scala._
import org.apache.flink.table.functions.AggregateFunction
import org.apache.flink.types.Row

object AggFunctionTest {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val settings = EnvironmentSettings.newInstance()
      .useBlinkPlanner()
      .inStreamingMode()
      .build()

    val tableEnv = StreamTableEnvironment.create(env, settings)


    val inputPath = "src/main/resources/SensorReading"
    val inputStream = env.readTextFile(inputPath)

    //转换成样例类类型
    val dataStream = inputStream.map(
      data => {
        val arr = data.split(",")
        SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
      }
      //选自字段作为时间戳
    ).assignTimestampsAndWatermarks(
      new BoundedOutOfOrdernessTimestampExtractor[SensorReading](Time.seconds(1)) {
        override def extractTimestamp(t: SensorReading) = t.timeStamp
      })
    val sensorTable = tableEnv.fromDataStream(dataStream
      , 'id, 'temperature, 'timeStamp.rowtime as 'ts)

    //Table API
    val avgTemp = new AvgTemp()
    val resultTable = sensorTable
      .groupBy('id)
      .aggregate(avgTemp('temperature) as 'avg_temp)
      .select('id, 'avg_temp)

    //sql
    //需要先注册
    tableEnv.createTemporaryView("sensor", sensorTable)
    tableEnv.registerFunction("avgTemp", avgTemp)
    val resultSqlTable = tableEnv.sqlQuery(
      """
        |select id,avgTemp(temperature) from sensor
        |group by id
        |""".stripMargin
    )

    resultTable.toRetractStream[Row].print("table")
    resultSqlTable.toRetractStream[Row].print("sql")
    env.execute()


  }
}

//定义一个类，用于表示聚合的状态
class AvgTempAcc {
  var sum: Double = 0.0
  var count: Int = 0
}

/**
 * 自定义一个聚合函数，求每个传感器的平均温度
 * 两个参数，一个是输入的数据，一个是保存在agg里面的状态
 */
class AvgTemp extends AggregateFunction[Double, AvgTempAcc] {
  //状态怎么计算
  override def getValue(acc: AvgTempAcc): Double = acc.sum / acc.count

  //创建初始的状态值
  override def createAccumulator(): AvgTempAcc = new AvgTempAcc

  //还要实现具体的处理计算函数
  def accumulate(acc: AvgTempAcc, temp: Double) = {
    acc.sum += temp
    acc.count += 1
  }
}